package com.xyh.sparkstreaming

import java.util.Properties

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.util.Random

object LogdataShow {
  def main(args: Array[String]): Unit = {
    // 发送Kafka Topic
    val props = new Properties()
    props.put("bootstrap.servers", "192.168.222.135:9092")
    props.put("acks", "1")
    props.put("retries", "3")
    props.put("key.serializer", classOf[StringSerializer].getName)
    props.put("value.serializer", classOf[StringSerializer].getName)
    val producer = new KafkaProducer[String, String](props)

    val random = new Random()
    val allStatus = Array(
      "L", "A"
    )
    val allName = Array(
      "Lawg", "Aesrgf", "SRG", "Sdfbxcv","QQWe", "ZXC", "QSTB", "YIHB"
    )
    val allTerms = Array(
      "1", "2", "3", "4","5", "6", "7", "8"
    )
    val allSex = Array(
      "1", "0"
    )
    val allDate = Array(
       "20221219", "20021210","20070801", "20060928", "20040616"
    )
    val allID = Array(
      "205815", "205816","205817"
    )
    val allStore = Array(
      "0", "1","2", "3", "4","5", "6", "7", "8","9","10"
    )
    //ID, name，sex1 0，date 8，num 10，term ，s-status,store
    while (true) {
      val classID : String = allID(random.nextInt(allID.length))
      val Name: String = allName(random.nextInt(allName.length))
      val Sex: String = allSex(random.nextInt(allSex.length))
      val Date: String = allDate(random.nextInt(allDate.length))
      val num: String = "1%09d".format(random.nextInt(10))
      val Status: String = allStatus(random.nextInt(allStatus.length))
      val Terms: String = allTerms(random.nextInt(allTerms.length))
      val Store: String =allStore(random.nextInt(allStore.length))
      //  val callDuration = if ("success".equals(callStatus)) (1 + random.nextInt(10)) * 1000L else 0L

      // 随机产生一条基站日志数据
      val stationLog: StationLog = StationLog(
        classID,
        Name,
        Sex,
        Date,
        num,
        Status,
        Terms,
        Store,


      )
//      println(stationLog.toString)
      Thread.sleep(100 + random.nextInt(100))
      //Thread.sleep(5000)
//      Thread.sleep(2000)
      val value: String = JSON.toJSONString(stationLog, new SerializeConfig(true))
      println(value)

      val record = new ProducerRecord[String, String]("stulnfo", value)
      producer.send(record)
      val record2 = new ProducerRecord[String, String]("result", stationLog.toString)
      producer.send(record2)
    }

    producer.close() // 关闭连接
  }

  /**
   * 基站通话日志数据
   */
  case class data(
                         classID: String, //ID
                         Name:String,//name
                         Sex: String, //sex
                         Date: String, //date
                         num: String, //学号
                         Status: String, //状态
                         Term: String, //学期
                         Store: String, //绩点
                       ){}
  case class StationLog(
                         classID: String, //ID
                         Name:String,//name
                         Sex: String, //sex
                         Date: String, //date
                         num: String, //学号
                         Status: String, //状态
                         Term: String, //学期
                         Store: String, //绩点
                       ) {
    override def toString: String = {
      s"$classID,$Name,$Sex,$Date,$num,$Status,$Term,$Store"
    }
  }




}



